/* 
 * Copyright (c), nFractals Ltd. All Rights Reserved.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at 
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */


package org.parade.distributed;

import java.util.Map;
import java.util.concurrent.locks.Lock;

import org.parade.Parade;
import org.parade.driver.Driver;

public class AtomicLong extends Number{

	private Driver driver;
	private Id id;
	private Map dataMap;
	private Lock dataLock;

	
	/**
	 * Create AtomicLong using driver provided by singleton Parade object
	 * 
	 * @param id		Cluster wide unique object id
	 */
	public AtomicLong(Id id){
		initialize(Parade.getDriver(), id, 0);
	}
	
	
	/**
	 * Create AtomicLong using driver provided by singleton Parade object
	 * 
	 * @param id		Cluster wide unique object id
	 * @param initialValue	initial value
	 */
	public AtomicLong(Id id, long initialValue){
		initialize(Parade.getDriver(), id, initialValue);
	}
	
	
	/**
	 * Create AtomicLong using given driver
	 * 
	 * @param driver	Distributed driver to use
	 * @param id		Cluster wide unique object id
	 */
	public AtomicLong(Driver driver, Id id){
		initialize(driver, id, 0);
	}
	
	/**
	 * Create AtomicLong using given driver
	 * 
	 * @param driver	Distributed driver to use
	 * @param id		Cluster wide unique object id
	 * @param initialValue	initial value
	 */
	public AtomicLong(Driver driver, Id id, long initialValue){
		initialize(driver, id, initialValue);
	}
	
	
	private void initialize(Driver driver, Id id, long initialValue){
		this.driver = driver;
		this.id = id;
		
		this.dataMap = driver.getMap(new Id(id.toString()+"-dataMap"));
		this.dataLock = driver.getLock(new Id(id.toString()+"-dataLock"));
		
		putValue(new Long(initialValue));
	}
	
	private void putValue(Long v){
		dataMap.put("DATA", v);
	}
	
	private Long getValue() throws Exception{
		Long v = (Long) dataMap.get("DATA");
		return v;
	}
	
	public double doubleValue() {
		dataLock.lock();
		try{
			return getValue().doubleValue();
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}

	@Override
	public float floatValue() {
		dataLock.lock();
		try{
			return getValue().floatValue();
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}

	@Override
	public int intValue() {
		dataLock.lock();
		try{
			return getValue().intValue();
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}

	@Override
	public long longValue() {
		dataLock.lock();
		try{
			return getValue().longValue();
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}

	public long addAndGet(long delta){
		dataLock.lock();
		try{
			long ret = getValue().longValue();
			ret+=delta;
			putValue(new Long(ret));
			return ret;
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}
	
	public boolean compareAndSet(long expect, long update){
		dataLock.lock();
		try{
			long ret = getValue().longValue();
			if(ret==expect){
				putValue(new Long(update));
				return true;
			} else {
				return false;
			}
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}
	
	public long getAndAdd(long delta){
		dataLock.lock();
		try{
			long ret = getValue().longValue();
			putValue(new Long(ret+delta));
			return ret;
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}
	
	public long getAndSet(long newValue){
		dataLock.lock();
		try{
			long ret = getValue().longValue();
			putValue(new Long(newValue));
			return ret;
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}
	
	public long get(){
		dataLock.lock();
		try{
			long ret = getValue().longValue();
			return ret;
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}
	
	public void set(long newValue){
		dataLock.lock();
		try{
			putValue(new Long(newValue));
		}catch(Exception e){
			throw new RuntimeException(e);
		} finally{
			dataLock.unlock();
		}
	}
	
	public long decrementAndGet(){
		return addAndGet(-1);
	}
	
	public long getAndDecrement(){
		return getAndAdd(-1);
	}
	
	public long getAndIncrement(){
		return getAndAdd(1);
	}
	
	public long incrementAndGet(){
		return addAndGet(1);
	}
	
	public String toString(){
		return Long.toString(get());
	}
	

}
